//Chao Wang
//chw412@lehigh.edu

//this documents corresponds to the Question#4
//of the assignment due on Dec. 3th, 2012
  

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <ctype.h>
#include <sys/stat.h>
#include <dirent.h>

#include <iostream>
#include <string.h>
#include <assert.h>
#include <map>
#include <fstream>
#include <set>
#include <vector>
#include <algorithm>
//#include <functional>

#include "mpi.h"

using namespace std;

extern char *optarg;
extern int optind;
extern int optopt;
extern int opterr;

//we assign a default value for N
int N = 3;

//File names containing URLs
char web_resource_file[1024] = "";

//This will contain the destination folder storing files from the web URL
char dest_dir[1024];


int rank; //the rank of me(i.e., this process)
int size; //the number of processes in all


int to = 0;   //the destination rank
int from = 0; //the source rank
int send_or_receive_count = 0;//the count of elements to be sent/received
int tag = 0;  //the tag of the message

int st_count, st_source, st_tag; //the information of the message that actually recieved.
MPI::Status status; //the status of the received message

char buffer[1<<14]; //the buffer used to store the file names send from rank 0 to all the ranks

class Word
{
public:
	string word;
	int count;
public:
	Word(void) : count(1) {}

	bool operator < (const Word & word) const
	{
		if(this->count > word.count)
			return true;
		else if (this->count < word.count)
			return false;
		else	
			return (this->word).compare(word.word) <= 0 ? true : false;
	}
};

//This is the dictionary we maintain in order to store the map between each distinct word and its number of occurance in a file.
class Dict
{
public:
	map<string, Word> dict;
	string filename;
public:
	Dict(string fn) : filename(fn) {}
	void Add(char * word);
	void MostCommonWords(int n = N);
	~Dict();
};

void Dict::MostCommonWords(int n) //store the words that occur most frequently in a set and print them
{
	set <Word> words;
	strcpy(buffer, "");//generate a buffere to store most frequent words (the parameter of -n) in it

	for(map<string, Word>::iterator i = dict.begin(); i != dict.end(); i++)
		words.insert(i->second);//insert it in the set 

	for(set <Word>::iterator i = words.begin(); i != words.end(); ++i)//
	{
		if(--n >= 0)
		{
			char temp[1024];
			//Each entry of the message contains three parts, the belonged filename, the word, and
			//its count
			//construct the message to be sent
			sprintf(temp, "%s,%s,%d\n", filename.c_str(), i->word.c_str(), i->count);
			strcat(buffer, temp);
		}
		else
			break;
	}

}


//This funciton add a word into a dictionary
void Dict::Add(char * word)
{
	//if this word is empty, do nothing
	if(strcmp(word, "") == 0)
		return;
	
	//change the word into lowercase word
  	for (int i = 0; word[i]; i++)
	    word[i] = tolower(tolower(word[i]));

	//find the place where the word is located in the dictionary.
	//if this word is not found, insert the new word, esle we increment its frequency
	string word_str(word);
	map<string, Word>::iterator map_ptr;
	map_ptr = dict.find(word_str);
	
	if(map_ptr != dict.end())
	{
		Word & word = dict[word_str];
		word.count++;
	}
	else
	{
		Word * word = new Word();
		word->word = word_str;
		dict[word_str] = *word;
	}
}

//deconstructing function
Dict::~Dict()
{
	for(map<string, Word>::iterator wi = dict.begin(); wi != dict.end(); wi++)
		delete &(wi->second);
}


//for each url, we process to download the file
void process_this_file(string url)
{
	//constuct the command to download
	strcpy(buffer, "wget -q ");
	strcat(buffer, url.c_str());
	strcat(buffer, " -O ");
	
	string filename(dest_dir);
	std::replace(url.begin(), url.end(), '/', '-');
	filename += url;

	strcat(buffer, filename.c_str());
	//download the file from url given, and store it to the directory /tmp/chw412/
	system(buffer);

	//open the downloaded file
	fstream file(filename.c_str(), ios::in);
	if(!file.is_open())
	{
		cerr<<"file "<<filename<<" cannot be open!";
		return;
	}
	
	//create a new dictionary
	Dict * pdict = new Dict(filename);


	//process each line, from which we split words according to the punctuaions.
	string buf;
	while(getline(file, buf))
	{
		char * c_buffer = new char [buf.size() + 1];
		strcpy(c_buffer, buf.c_str());

		char * word = strtok(c_buffer, " \t\n\v\f\r.?!:;-()[]'\"/,");

		while(word)
		{
			pdict->Add(word);
			word = strtok(NULL, " \t\n\v\f\r.?!:;-()[]'\"/,");
		}

		delete [] c_buffer;
	}
	
	//calcuate the most N frequent occuring words in this dictionary
	pdict->MostCommonWords(N);

	delete pdict;
	file.close();
}

//Send the reslt to the rank 0
void SendResult(char * buffer)
{
	assert(buffer);

	to = 0;

	tag = st_tag;
	send_or_receive_count = strlen(buffer) + 1;
	MPI::COMM_WORLD.Send(buffer,send_or_receive_count, MPI::CHAR, to, tag);
}

//tell other processors to die
void TellOthersToDie(void)
{
	for(int i = 1; i < size; i ++)
	{
		MPI::COMM_WORLD.Send("YOU_CAN_DIE_NOW", 16, MPI::CHAR, i, 0);
	}
}


//the main program 
int main(int argc, char ** argv)
{
	MPI::Init(argc, argv);

	//get the rank of this processor, as well as the number of processors in this communcation world.
	rank = MPI::COMM_WORLD.Get_rank();
	size = MPI::COMM_WORLD.Get_size();

	optind = 1;
	opterr = 0;
	int result;

	//using the getopt function to handle options
	while((result = getopt(argc, argv, "l:n:")) != -1)
	{
		switch((char)result)
		{
			case 'l':
				strcpy(web_resource_file, optarg);
				break;
			case 'n':
				N = atoi(optarg); 
				break;
			case '?':
				if (optopt == 'n' || optopt == 'l')
					fprintf(stderr, "Option -%c requires an argument.\n", optopt);
				else if (isprint (optopt))
					fprintf(stderr, "Unknown option '-%c'.\n", optopt);
				else
					fprintf(stderr, "Unkown option character '\\x%x'.\n", optopt);
				return 1;
			default:
				abort();

		}
	}

	//Here we sucessfully get the value of N, which indicates the number of the most common words we need to output.
	//printf("N = %d\n", N);
	if(optind < argc) //there is no file(s) that need to be processed
		fprintf(stderr, "You do not have to designate files as parameters! Files designated are automatically omitted.\n");

	//test if the directory is empty string,
	//if it is, all processes exit.
	if(rank == 0 && strcmp(web_resource_file, "") == 0)
	{
		fprintf(stderr, "You need to assign a web-resource-file using -l option!\n");
		exit(0);
	}

	struct stat st;


	//the destionation folder where we need to store the downloaded files
	strcpy(dest_dir, "/tmp/chw412/");


	if(stat(dest_dir, &st) != 0) 	//the destination directory does not exist
	{				//create new directory
		mkdir(dest_dir, 0777);
	}
	else
	{
		//remove all the non-hidden files in this directory
		struct dirent * entryptr;
		DIR * dirptr;
		dirptr = opendir(dest_dir);
		while((entryptr = readdir(dirptr)) != NULL)
		{
			if(entryptr->d_name[0] == '.')
				continue;
			
			strcpy(buffer, dest_dir);
			strcat(buffer, entryptr->d_name);
			unlink(buffer);
		}
		closedir(dirptr);
	}


	if(rank == 0) //I am the leading rank
	{
		double start_time = 0.0f, end_time = 0.0f;
		start_time = MPI::Wtime();
		//I should read this URL file
		
		vector<string> urls;
		
		//open the web file that contain urls
		fstream wrf(web_resource_file, ios::in);
		string buf;

		//for each line, get one url
		while(getline(wrf, buf))
			urls.push_back(buf);

		/*
		vector<string>::iterator i;
		for(i = urls.begin(); i != urls.end(); i++)
			cout << *i << endl;
		cout << endl;
*/
		
		unsigned int round = 0;

		//the process of sending the work to all other processors
		//we split the files into different groups
		//each time we send a group of files to all the processors
                while(round * size < urls.size()) 
		{
			for(unsigned int i = round * size; i < urls.size() && i < (round + 1) * size; i ++) 
			{
				//Send the URL as a MPI message
				to = i % size;
				send_or_receive_count = urls[i].length() + 1;
				tag = i;

				MPI::COMM_WORLD.Send(urls[i].c_str(), send_or_receive_count, MPI::CHAR, to, tag);
			//	fprintf(stdout, "rank %d: files %s(tag %d) sent to %d.\n\n", rank, urls[i].c_str(), tag, to);
			}

			//since rank 0 will always apear in every round, rank 0 therefore tries to finish its own
			//job on processing files. 
			from = MPI::ANY_SOURCE;
			send_or_receive_count = (1<<14);
			tag = MPI::ANY_TAG;
			MPI::COMM_WORLD.Recv(buffer, send_or_receive_count, MPI::CHAR, from, tag, status);

			st_count = status.Get_count(MPI::CHAR);
			st_source = status.Get_source();
			st_tag = status.Get_tag();

			//processor 0 processes the file and send the result to itself
			process_this_file(string(buffer));
			SendResult(buffer);
                       
			//recieve all the results from all other machines.
			for(unsigned int i = round * size; i < urls.size() && i < (round + 1) * size; i ++)
			{
				//Receive the MPI message 
				from = MPI::ANY_SOURCE;
				send_or_receive_count = (1<<14);
				tag = MPI::ANY_TAG;
				
				MPI::COMM_WORLD.Recv(buffer, send_or_receive_count, MPI::CHAR, from, tag, status);
				st_count = status.Get_count(MPI::CHAR);
				st_source = status.Get_source();
				st_tag = status.Get_tag();

//				fprintf(stdout, "rank %d: received content:\n%s\n", rank, buffer);
				fprintf(stdout, "%s\n", buffer);
			}

			round++;
		}
		
		TellOthersToDie();


		//we need this timer for recording the time used in our model.
		end_time = MPI::Wtime();
		fprintf(stderr, "%d\t%lf\n", size, end_time - start_time);
	}
	else
	{

		//this loop makes sure that receives will receive messages until it eventually obtains a message
		//telling it to die.

		while(1)
		{
			from = MPI::ANY_SOURCE;
			send_or_receive_count = (1<<14);
			tag = MPI::ANY_TAG;
			MPI::COMM_WORLD.Recv(buffer, send_or_receive_count, MPI::CHAR, from, tag, status);
			st_count = status.Get_count(MPI::CHAR);
			st_source = status.Get_source();
			st_tag = status.Get_tag();
			
			if(strcmp(buffer, "YOU_CAN_DIE_NOW") == 0)
				break;

			process_this_file(string(buffer));
			SendResult(buffer);
		}


	}

	MPI::Finalize();
	return 0;
}

